package storm;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import java.util.*;
/**
 * @author Hanhan_Jun
 * @date 2021年12月12日 15:12
 */
public class Bolt1_1 extends BaseBasicBolt{
    //连接redis，ip地址是本机
    public static JRedisUtil jredis =new JRedisUtil("10.86.184.170",6379);

    public void execute(Tuple input, BasicOutputCollector collector) {
        try {
            String book_name = input.getStringByField("book_name");
            int book_num = input.getIntegerByField("book_num");
            String book_press=input.getStringByField("book_press");
            String book_type=input.getStringByField("book_type");
            Double book_price = input.getDoubleByField("book_price");

            //有序集合
            jredis.zincrby("Top10Books",book_num,book_name);
            jredis.zincrby("Top10Press",book_num,book_press);
            jredis.zincrby("Top10BookType",book_num,book_type);
            jredis.incrBy("TotalBooksNum", (long) book_num);
            jredis.zincrby("price",book_num,book_price.toString());
            System.out.println("Done111111111111111111111111111111111111");
        }catch (Exception e){
            e.printStackTrace();
        }

    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {}
}
